package com.zcs.thread.pool;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 自定义创建线程池
 *
 * @author zcs
 */
public class MyThreadPoolExecutor implements Executor {
    /**
     * 核心流程：
     * 线程池中有N个工作线程
     * 把任务提交给线程池运行
     * 如果线程池已满，把任务放入队列
     * 最后当有空闲时，获取队列中任务来执行
     */

    /**
     * 记录线程池中线程数量
     */
    private final AtomicInteger ctl = new AtomicInteger(0);
    /**
     * 核心线程数
     */
    private volatile int corePoolSize;
    /**
     * 最大线程数
     */
    private volatile int maximumPoolSize;
    /**
     * 阻塞队列
     */
    private final BlockingQueue<Runnable> workQueue;

    public MyThreadPoolExecutor(int corePoolSize, int maximumPoolSize, BlockingQueue<Runnable> workQueue) {
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
    }

    @Override
    public void execute(Runnable command) {
        //工作线程数
        int c = ctl.get();
        //小于核心线程数
        if (c < corePoolSize) {
            //创建线程运行这个任务
            if (!addWorker(command)) {
                //任务添加失败, 执行拒绝策略
                reject();
            }
            return;
        }
        //任务队列添加任务
        if (!workQueue.offer(command)) {
            //任务队列满，尝试启动线程添加任务
            if (!addWorker(command)) {
                reject();
            }
        }
    }

    /**
     * 饱和拒绝
     */
    private void reject() {
        //直接抛出异常
        throw new RuntimeException("Can not execute!ctl.count："
                + ctl.get() + "workQueue size：" + workQueue.size());
    }

    /**
     * 添加任务
     *
     * @param firstTask
     * @return
     */
    private boolean addWorker(Runnable firstTask) {
        if (ctl.get() >= maximumPoolSize) {
            return false;
        }
        Worker worker = new Worker(firstTask);
        //启动线程
        worker.thread.start();

        ctl.incrementAndGet();
        return true;
    }


    /**
     * 线程池工作线程包装类
     * <p>
     * Worker是ThreadPoolexecutor的内部类，主要用于维持线程执行任务的中断控制状态。
     */
    private final class Worker implements Runnable {

        final Thread thread;
        Runnable firstTask;

        public Worker(Runnable firstTask) {
            this.thread = new Thread(this);
            this.firstTask = firstTask;
        }

        @Override
        public void run() {
            Runnable task = firstTask;
            try {
                //执行任务
                while (task != null || (task = getTask()) != null) {
                    task.run();
                    //线程池已满，跳出循环
                    if (ctl.get() > maximumPoolSize) {
                        break;
                    }
                    task = null;
                }
            } finally {
                //工作线程数增加
                ctl.decrementAndGet();
            }
        }

        /**
         * 从队列中获取任务
         *
         * @return
         */
        private Runnable getTask() {
            for (; ; ) {
                try {
                    return workQueue.take();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    //测试
    public static void main(String[] args) {
        MyThreadPoolExecutor myThreadPoolExecutor = new MyThreadPoolExecutor(2, 2,
                new ArrayBlockingQueue<Runnable>(10));
        for (int i = 0; i < 10; i++) {
            int taskNum = i;
            myThreadPoolExecutor.execute(() -> {
                try {
                    Thread.sleep(1500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("任务编号：" + taskNum);
            });
        }
    }
}
